package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.SecStockHbaseFunction;
import cn.itcast.function.SecStockWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.sink.SinkHbase;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Data 2022/5/13
 * @Describtion 秒级行情
 * @Author haozhi
 * @Version 1.0
 */
public class StockSecTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {

        /**
         * 1.每5s生成一条行情数据(事件时间最大）
         * 2.使用批量插入，需要分装到List<put>
         * 3.窗口时间5s
         */
        /**
         * 开发步骤
         *  1.数据分组
         *  2.划分时间窗口
         *  3.新建个股数据写入bean对象
         *  4.秒级窗口函数业务处理
         *  5.数据写入操作
         *    * 封装ListPuts
         *    * 数据写入
         */
        //todo 1.数据分组
        //根据个股代码进行分组
        waterData.keyBy(new KeyFunction())
                .timeWindow(Time.seconds(5))
                //秒级窗口函数业务处理
                .apply(new SecStockWindowFunction())
                .timeWindowAll(Time.seconds(5))//不分组划分时间窗口
                .apply(new SecStockHbaseFunction()) //收集的是5s窗口之内的所有数据
                .addSink(new SinkHbase(QuotConfig.config.getProperty("stock.hbase.table.name")));
        //todo 2.划分时间窗口
        //todo 3.新建个股数据写入bean对象
        //todo 4.秒级窗口函数业务处理
        //todo 5.数据写入操作
        //todo   * 封装ListPuts
        //todo   * 数据写入
    }
}
